package com.tool.rabbit;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.log.StaticLog;
import com.rabbitmq.client.*;
import ntd.util.ComConfig;
import ntd.util.fun.ComFun;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @author: liuwenju
 * @date: 2023/8/7 9:51
 * @version: 1.0
 */
public class ToolRabbitMQ {

    private static Map<String, Channel> channelMap = new HashMap<>();

    private static ConnectionFactory factory = new ConnectionFactory();

    static {
        factory.setUsername(ComConfig.RABBIT_MQ_USERNAME);
        factory.setPassword(ComConfig.RABBIT_MQ_PASSWORD);
        factory.setHost(ComConfig.RABBIT_MQ_URL);
    }

//    public static void init_exchange(String queue) {
//        Channel channel = init_channel("push", queue);
//        // 4.创建交换机
//        try {
//            String exchage_key = queue + "_exchange_topic";
//            channel.exchangeDeclare(exchage_key, BuiltinExchangeType.TOPIC,true);
//            channel.queueBind(queue, exchage_key,queue + ".*");
//        } catch (IOException e) {
//            throw new RuntimeException(e);
//        }
//    }

    public static Channel init_channel(String type, String queue) {
        try {
            Channel channel = channelMap.get(type + queue);
            if (ObjectUtil.isNull(channel)) {
                // 创建连接
                Connection connection = factory.newConnection();
                // 创建通道
                channel = connection.createChannel();
                // 创建队列
                channel.queueDeclare(queue, false, false, false, null);
                // 4.创建交换机
//                channel.exchangeDeclare(queue + "_exchange_topic", BuiltinExchangeType.TOPIC,true);
                // 缓存通道
                channelMap.put(type + queue, channel);
                StaticLog.info("LJ创建新的RabbitMQ通道队列[{}]", type + queue);
            }
            return channel;
        } catch (IOException | TimeoutException e) {
            throw new RuntimeException(e);
        }
    }

    public static void push(String queue, String msg) {
        Channel push = init_channel("push", queue);
        try {
//            StaticLog.info("LJ队列[{}]推送消息[{}]", queue, msg);
            // MessageProperties.PERSISTENT_TEXT_PLAIN 持久化消息，传null为不需要持久化
            push.basicPublish("", queue, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static void on_msg(String queue, Consumer consumer) {
        Channel channel = init_channel("on_msg", queue);
        try {
            StaticLog.info("LJ开始监听队列[{}]", queue);
            channel.basicConsume(queue,true, consumer);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public static void on_msg(String queue, ComFun.FunInNoException<String> inNoException) {
        Channel channel = init_channel("on_msg", queue);
        try {
            StaticLog.info("LJ开始监听队列[{}]", queue);
            channel.basicConsume(queue,true, new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body, "UTF-8");
                    inNoException.accept(message);
                }
            });
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
